// REQUIRES THE BERKELEY DB, SO NOT BUILT BY DEFAULT

using namespace std;

#define __STDC_FORMAT_MACROS

// #define DEBUG_TRANS

#include <sys/stat.h>
#include <sys/types.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <inttypes.h>
#include <errno.h>
#include <assert.h>
#include <exception>
#include <db.h>
#include <string>
#include <map>
#include "pwext.hpp"

// [[ex06ca]]
// DEBUG descr The sequence class for generation of the unique object keys.
class sequence
{
public:
	typedef uint64_t key_t;

	sequence() :
		sequence_(0)
	{ }

	// generate the next sequence,
	// guaranteed to never generate 0
	key_t next()
	{
		pw::lockmutex lm(seq_mutex_);
		++sequence_;
		// 64 bit should be enough to never wrap
		assert(sequence_ != 0); 
		return sequence_;
	}

protected:
	// sequence for generation of the keys
	pw::pmutex seq_mutex_;
	key_t sequence_;
};
// END

// [[ex06ci]]
// DEBUG descr The transaction exception class.
class trans_exception : public exception
{
public:
	trans_exception(int err, const char *s) :
		trans_err_(err), msg_(s)
	{ 
		if (err != 0) {
			msg_.append(": ");
			msg_.append(db_strerror(err));
		}
	}

	~trans_exception() throw()
	{ 
		if (trans_err_ == DB_LOCK_DEADLOCK)
			usleep(10);
	}

	virtual const char *what() const throw()
	{
		return msg_.c_str();
	}

	int trans_err_;
	std::string msg_;
};
// END

class transaction;

// [[ex06cb]]
// DEBUG descr Declaration of the database wrapper class.
class database
{
	friend class transaction;
public:
	database() :
		env_(0), table_(0), valid_(false)
	{
	}

	~database()
	{
		close();
	}

	// open the database instance
	// returns 0 on success or error code
	int open(const char *dirname, int page_size, 
		int max_pages, int max_trans);

	// close the database instance
	void close();

public:
	sequence seq_;

protected:
	// BDB environment
	DB_ENV *env_; 
	// the sole table in it
	DB *table_; 
	// flag: opened successfully
	bool valid_; 
	// name of the directory in memory filesystem
	std::string dirname_; 
};
// END


// [[ex06cg]]
// DEBUG descr Declaration of the transaction class.
class transaction 
{
	friend class tmptr_base;
public:
	typedef sequence::key_t key_t;

	transaction(database *db);
	// must be committed, rolled back or failed before
	// destruction; commits by default, unless rolled
	// back
	~transaction();

	// explicitly commit the changes;
	// rolls back and throws on errors
	void commit(); 
	// roll back the transaction; return 0 on success,
	// -1 if failed previously, or error code
	int rollback(); 

	// check whether the transaction has failed
	bool has_failed() const { return commitfailed_; }

protected:
	// to be called from tmptr_base, throw exceptions
	// on failure
	typedef void destructor_func(void *obj);
	void *get_cached(key_t k);
	void *get_mod_cached(key_t k);
	void get_into(key_t k, void *newobj, size_t objsz, 
		destructor_func *dest, destructor_func *compdest);
	key_t add_new(void *newobj, size_t objsz, 
		destructor_func *dest, destructor_func *compdest);
	void set_modified(key_t k, void *newobj);
	void destroy(key_t k);
	void uncache(key_t k);

protected:
	// DEBUG {
	// The things that need to keep track of can be
	// divided into states:
	// * "clean" objects that were read-only: on
	//   uncaching, commit or rollback destroy
	//   shallowly, remove from map
	// * "dirty" objects that were read and modified:
	//   on commit or uncaching attempt to update the
	//   copy in the database, on succesful commit
	//   destroy shallowly the new copy, deeply the old
	//   copy; on rollback destroy shallowly the old
	//   copy, deeply the new copy
	// * newly added objects: on commit or uncaching
	//   attempt to insert in the database, on succesful
	//   commit destroy shallowly the new copy; on
	//   rollback destroy deeply the new copy
	// * deleted objects: immediately attempt to
	//   delete in the database, on succesful commit
	//   destroy deeply the old copy; on rollback
	//   destroy shallowly the old copy
	//
	// The possible transitions between states are:
	// clean->dirty:
	//   create the new copy by deep-copying
	// clean->deleted:
	//   nothing to do (besides the delete in the
	//   database)
	// dirty->deleted:
	//   immediately destroy deeply the new copy
	// new->deleted:
	//   immediately destroy deeply the new copy
	// DEBUG }

	// information about the cached objects
	class obj_info {
	public:
		enum state {
			CLEAN,
			DIRTY,
			ADDED,
			DELETED
		};

		obj_info()
		{ }
		obj_info(state st, bool cur, void *oldc, void *newc, 
				size_t sz, destructor_func *shallowd,
				destructor_func *deepd) : 
			state_(st), dbcurrent_(cur), oldcopy_(oldc),
			newcopy_(newc), objsz_(sz),
			shallow_dest_(shallowd), deep_dest_(deepd)
		{ }

		state state_;
		// flag: the database contains a current copy
		bool dbcurrent_; 
		// unmodified copy of the object
		void *oldcopy_; 
		// modified copy of the object
		void *newcopy_; 
		// object size
		size_t objsz_; 
		// wrapper of C++ destructor function for this
		// object
		destructor_func *shallow_dest_; 
		// wrapper of destroy() function
		destructor_func *deep_dest_; 
	};
	typedef std::map<key_t, obj_info> obj_map;

	// write the modifications of the object (if any)
	// back to the database returns 0 on success or an
	// error code
	int writeback(obj_map::iterator &it);

	database *db_;
	DB_TXN *txn_; // the current BDB transaction

	obj_map cache_; // map of unmodified objects

	// key of the record for update
	DBT key_; 
	// data of the record for update
	DBT data_; 
	// flag: some operation in transaction has failed
	bool commitfailed_; 
};
// END

// an empty class, used as a token for the constructor
// [[ex06cd]]
// DEBUG descr The empty token class used to differentiate the constructors.
class tmptr_token
{
};
// END

// [[ex06ce]]
// DEBUG descr Declaration of the transaction-managed pointer class.
class tmptr_base
{
	friend class transaction;
public:
	typedef sequence::key_t key_t;

	tmptr_base()
	{ }

	// for easy initialization to 0
	tmptr_base(key_t k) :
		k_(k)
	{ }

	// for convenience of assigning and comparison to 0
	tmptr_base &operator=(key_t k)
	{
		// no arbitrary values to be assigned
		assert(k == 0); 
		k_ = k;
		return *this;
	}
	bool operator==(key_t k)
	{
		return k_ == k;
	}
	bool operator!=(key_t k)
	{
		return k_ != k;
	}

	// comparison of pointers
	bool operator==(const tmptr_base &ob) const
	{
		return k_ == ob.k_;
	}
	bool operator!=(const tmptr_base &ob) const
	{
		return k_ != ob.k_;
	}

	// notify the transaction that the cached copy can
	// be discarded (if modified, it will be written
	// to the database first)
	void uncache(transaction &trans)
	{
		trans.uncache(k_);
	}

	// the rest of object control methods are in the
	// template
protected:
	// wrappers for the transaction functions, since
	// C++ "friend" does not allow the direct access
	// for the derived classes of a friend
	typedef void destructor_func(void *obj);
	inline void *get_cached(transaction &trans) const
	{
		return trans.get_cached(k_);
	}
	inline void *get_mod_cached(transaction &trans) const
	{
		return trans.get_mod_cached(k_);
	}
	inline void get_into(transaction &trans, 
		void *newobj, size_t objsz, destructor_func *dest,
		destructor_func *compdest) const
	{
		trans.get_into(k_, newobj, objsz, dest, compdest);
	}
	inline void add_new(transaction &trans, void *newobj, 
		size_t objsz, destructor_func *dest, 
		destructor_func *compdest)
	{
		k_ = trans.add_new(newobj, objsz, dest, compdest);
	}
	inline void set_modified(transaction &trans, 
		void *newobj) const
	{
		trans.set_modified(k_, newobj);
	}
	inline void destroy(transaction &trans) const
	{
		trans.destroy(k_);
	}
	inline void uncache(transaction &trans) const
	{
		trans.uncache(k_);
	}

	key_t k_; // the actual value
};
// END

// The objects of class T must have some peculiar properties :
// 1. Their destuctor must be completely empty, not including even 
// the auto-generated code (this means that they can't inherit
// from classes with non-empty destructors, nor include them
// directly as members, instead everything must be pointers).
// This is because after the object's contents is copied to the
// database, its empty shell needs to be destroyed without 
// disturbing any members.
// 2. The real destruction of all contents must happen in the 
// methods destroy_tmptrs(transaction &t) and destroy(). 
// Destroy_tmptrs() destroys any references through tmptr<>,
// destroy() destroys the rest of the contents.
// 3. The constructor T(tmptr_token &) must be trivial,
// i.e. possibly setting some default values inside object's own
// memory but not creating any other objects; it will be used 
// to create the object "shell" before restoring its contents
// from the database.
// 4. Whenever an object is copied, it must be copied "deeply",
// to the full depth, except the references to tmptr. The
// copy constructor and assignment must be deep.

// [[ex06cf]]
// DEBUG descr Definition of the transaction-managed pointer methods.
template <class T>
class tmptr : public tmptr_base
{
public:
	// for easy initialization to 0
	tmptr(key_t k) :
		tmptr_base(k)
	{ }
	tmptr()
	{ }

	// get the underlying object, for reading only;
	// the transaction keeps owning it;
	// throw trans_exception on failure
	const T *get(transaction &trans) const
	{
		void *obj = tmptr_base::get_cached(trans);
		if (obj)
			return (T *)obj;
		T* newobj = new T(*(tmptr_token *)0);
		try {
			tmptr_base::get_into(trans, (void *)newobj, 
				sizeof(T), &call_cpp_destructor, 
				&call_destroy_and_destructor);
			return newobj;
		} catch (...) {
			delete newobj;
			throw;
		}
	}
	// get the underlying object, for a pending
	// modification;
	// the transaction keeps owning it;
	// throw trans_exception on failure
	T *get_mod(transaction &trans) const
	{
		void *obj = tmptr_base::get_mod_cached(trans);
		if (obj)
			return (T *)obj;
		T *orig = (T *)get(trans);
		// deep-copy the contents
		T *newobj = new T(*orig);
		tmptr_base::set_modified(trans, newobj);
		return newobj;
	}
	// make this tmptr point to a newly created object
	// that will be
	// added to the database;
	// throw trans_exception on failure
	void add_new(transaction &trans, T *obj)
	{
		tmptr_base::add_new(trans, (void *)obj, sizeof(T), 
			&call_cpp_destructor, 
			&call_destroy_and_destructor);
	}
	// destroy the object, both in the database and
	// any instance previously returned by get() or
	// get_mod()
	// of the template;
	// throw trans_exception on failure
	void destroy(transaction &trans)
	{
		if (k_ != 0) {
			// first must bring the object into the
			// modified cache
			T *obj = get_mod(trans); 
			// then let the object destroy any nested tmptrs
			obj->destroy_tmptrs(trans); 
			// then update the database
			tmptr_base::destroy(trans); 
			// not strictly necessary but a convenience
			k_ = 0; 
		}
	}

protected:
	static void call_cpp_destructor(void *obj)
	{
#ifdef DEBUG_TRANS
		fprintf(stderr, "Shallow destructor %p.\n", obj);
#endif
		delete (T *)obj;
	}
	static void call_destroy_and_destructor(void *obj)
	{
#ifdef DEBUG_TRANS
		fprintf(stderr, "Deep destructor %p.\n", obj);
#endif
		((T *)obj)->destroy();
		delete (T *)obj;
	}
};
// END

// -------------- database ------------------------------

// [[ex06cc]]
// DEBUG descr Definition of the database wrapper methods.
int database::open(const char *dirname, 
	int page_size, int max_pages, int max_trans)
{
	int res;

	// in case if it was already open, discard the old
	// instance
	close(); 

	if (mkdir(dirname, 0700) != 0)
		return errno;

	dirname_ = dirname;

	res = db_env_create(&env_, 0);
	if (res != 0) return res;
	env_->set_flags(env_, DB_AUTO_COMMIT, 1 );
	if (res != 0) return res;
	env_->log_set_config(env_, DB_LOG_AUTO_REMOVE, 1);
	if (res != 0) return res;
	env_->set_lk_max_locks(env_, max_pages);
	if (res != 0) return res;
	env_->set_lk_max_objects(env_, max_pages);
	if (res != 0) return res;
	env_->set_lk_max_lockers(env_, max_trans);
	if (res != 0) return res;
	env_->set_tx_max(env_, max_trans);
	if (res != 0) return res;
	env_->set_timeout(env_, 10, DB_SET_TXN_TIMEOUT);
	if (res != 0) return res;

	env_->open(env_, dirname,
		// Create the environment if it does not already
		// exist
		DB_CREATE | 
		// Initialize transactions
		DB_INIT_TXN | 
		// Initialize locking.
		DB_INIT_LOCK | 
		// Free-thread the env handle.
		DB_THREAD | 
		// Initialize the in-memory cache.
		DB_INIT_MPOOL | 
		// Used from only one process.
		DB_PRIVATE | 
		// Allow any paths for db files
		DB_USE_ENVIRON_ROOT 
		, 0);
	if (res != 0) return res;

	res = db_create(&table_, env_, 0);
	if (res != 0) return res;
	table_->set_pagesize(table_, page_size);
	if (res != 0) return res;
	table_->open(table_,
		NULL, // Txn pointer, autocommit
		"data.bdb", // File name
		"data", // Logical db name
		DB_BTREE, // Database type (using btree)
		// Open flags
		DB_CREATE | DB_EXCL | DB_AUTO_COMMIT | DB_THREAD, 
		0666); // File mode. Using defaults
	if (res != 0) return res;

	valid_ = true;
	return 0;
}

void database::close()
{
	valid_ = false;

	if (table_ != 0) {
		table_->close(table_, DB_NOSYNC);
		table_ = 0;
	}
	if (env_ != 0) {
		env_->close(env_, 0);
		env_ = 0;
	}

	if (!dirname_.empty()) {
		string cmd = "rm -rf '";
		cmd.append(dirname_);
		cmd.append("'");
		system(cmd.c_str());
		dirname_.clear();
	}
}
// END

// ----------- transaction --------------

// [[ex06ch]]
// DEBUG descr Definition of the transaction methods.
transaction::transaction(database *db) :
	db_(db), txn_(0), commitfailed_(false)
{
	assert (db_->valid_);

	int res = db_->env_->txn_begin(db_->env_, 
		NULL, &txn_, 0);
	if (res != 0) 
		commitfailed_ = true;

	memset(&key_, 0, sizeof key_);
	key_.flags = DB_DBT_USERMEM;
	key_.ulen = key_.size = sizeof(key_t);

	memset(&data_, 0, sizeof data_);
	data_.flags = DB_DBT_USERMEM;
}

transaction::~transaction()
{
	if (txn_) {
		// catch the end-of-block without a commit
		assert(commitfailed_); 
		rollback();
	}
}

void transaction::commit()
{
	int res;
	if(!txn_  || commitfailed_)
		throw trans_exception(0, "second commit attempt");

	// first write any updates
	for (obj_map::iterator it = cache_.begin(); 
			it != cache_.end(); ++it) {
		if (!it->second.dbcurrent_ && it->second.newcopy_) {
			res = writeback(it);
			if (res != 0) {
				commitfailed_ = true;
				rollback();
				throw trans_exception(res, "error in db put");
			}
		}
	}

	res = txn_->commit(txn_, DB_TXN_NOSYNC);
	txn_ = 0;

	if (res == 0) { 
		// success, apply the changes to memory management
		for (obj_map::iterator it = cache_.begin(); 
				it != cache_.end(); ++it) {
			switch (it->second.state_) {
			case obj_info::CLEAN:
				it->second.shallow_dest_(it->second.oldcopy_);
				break; 
			case obj_info::DIRTY:
				it->second.deep_dest_(it->second.oldcopy_);
				it->second.shallow_dest_(it->second.newcopy_);
				break;
			case obj_info::ADDED:
				it->second.shallow_dest_(it->second.newcopy_);
				break;
			case obj_info::DELETED:
				it->second.deep_dest_(it->second.oldcopy_);
				break;
			}
		}
		cache_.clear();
	} else {
		// commit failed, do the same way as on rollback
		rollback();
	}
	if (res != 0)
		throw trans_exception(res, "error in db commit");
#ifdef DEBUG_TRANS
	fprintf(stderr, "Committed.\n");
#endif
}

int transaction::rollback()
{
	for (obj_map::iterator it = cache_.begin(); 
			it != cache_.end(); ++it) {
		if (it->second.oldcopy_) {
			it->second.shallow_dest_(it->second.oldcopy_);
		}
		if (it->second.newcopy_) {
			it->second.deep_dest_(it->second.newcopy_);
		}
	}
	cache_.clear();

	if (!txn_)
		return 0;

	int res = txn_->abort(txn_);
	txn_ = 0;
	return res;
}

void *transaction::get_cached(key_t k)
{
	obj_map::iterator it = cache_.find(k);
	if (it == cache_.end()) {
		return 0;
	}

	void *ret;

	switch (it->second.state_) {
	case obj_info::CLEAN:
		ret = it->second.oldcopy_;
		break;
	case obj_info::DIRTY:
	case obj_info::ADDED:
		ret = it->second.newcopy_;
		break;
	case obj_info::DELETED:
		// a look-up of a deleted object is not good
		abort(); 
		return 0; 
	}

#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::get_cached key=%" 
		PRIu64 " at %p\n", k, ret);
#endif
	return ret;
}

void *transaction::get_mod_cached(key_t k)
{
	obj_map::iterator it = cache_.find(k);
	if (it == cache_.end()) {
		return 0;
	}
	// if it was clean or deleted, newcopy will be NULL
#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::get_mod_cached key=%" 
			PRIu64 " at %p\n", 
		k, it->second.newcopy_);
#endif
	if (it->second.newcopy_ && it->second.dbcurrent_)
		it->second.dbcurrent_ = false;
	return it->second.newcopy_;
}

void transaction::get_into(key_t k, void *newobj, 
	size_t objsz, destructor_func *shallowd, 
	destructor_func *deepd)
{
	key_.data = &k;
	data_.data = newobj;
	data_.ulen = data_.size = objsz;
	int res = db_->table_->get(db_->table_, txn_, 
		&key_, &data_, 0);
	if (res != 0) {
		commitfailed_ = true;
		assert(res != DB_NOTFOUND); // a fatal condition
		throw trans_exception(res, "error in db get");
	}
	if (data_.size != objsz) {
		fprintf(stderr, "transaction::get_into key=%" 
				PRIu64 " buffer %zu bytes, got %zu bytes\n",
			*(uint64_t*)key_.data, objsz, (size_t)data_.size);
		abort();
	}

	cache_[k] = obj_info(obj_info::CLEAN, true, 
		newobj, 0, objsz, shallowd, deepd);
#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::get_into key=%" 
			PRIu64 " at %p\n", k, newobj);
#endif
}

void transaction::set_modified(key_t k, void *newobj)
{
	obj_map::iterator it = cache_.find(k);
	// must be already in the cache
	assert(it != cache_.end()); 

	assert(it->second.state_ == obj_info::CLEAN);
	it->second.state_ = obj_info::DIRTY;
	it->second.dbcurrent_ = false;
	it->second.newcopy_ = newobj;

#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::set_modified key=%" 
			PRIu64 " from state %d at old %p new %p\n", 
		k, it->second.state_, it->second.oldcopy_, 
		it->second.newcopy_);
#endif
}

transaction::key_t transaction::add_new(
	void *newobj, size_t objsz, 
	destructor_func *shallowd, destructor_func *deepd)
{
	key_t k = db_->seq_.next();
	cache_[k] = obj_info(obj_info::ADDED, false, 
		0, newobj, objsz, shallowd, deepd);
#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::add_new key=%" 
			PRIu64 " at %p\n", k, newobj);
#endif
	return k;
}

void transaction::destroy(key_t k)
{
	obj_map::iterator it = cache_.find(k);
	// must be already in the cache
	assert(it != cache_.end()); 

#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::destroy key=%" 
			PRIu64 " from state %d at old %p new %p\n", 
		k, it->second.state_, it->second.oldcopy_, 
		it->second.newcopy_);
#endif

	switch (it->second.state_) {
	case obj_info::CLEAN:
		break; // no special transitions
	case obj_info::DIRTY:
		it->second.deep_dest_(it->second.newcopy_);
		it->second.newcopy_ = 0;
		break;
	case obj_info::ADDED:
		it->second.deep_dest_(it->second.newcopy_);
		it->second.newcopy_ = 0;
		cache_.erase(it);
		return;
	case obj_info::DELETED:
		abort(); // double-deletion is not good
		return; 
	}

	it->second.state_ = obj_info::DELETED;
	// will be immediately applied to the database
	it->second.dbcurrent_ = true; 

	key_.data = (void *)&it->first;
	int res = db_->table_->del(db_->table_, 
		txn_, &key_, 0);
	if (res != 0) {
		commitfailed_ = true;
		throw trans_exception(res, "error in db delete");
	}
}

void transaction::uncache(key_t k)
{
	obj_map::iterator it = cache_.find(k);
	if (it == cache_.end()) {
#ifdef DEBUG_TRANS
		fprintf(stderr, "transaction::uncache unknown key=%" 
				PRIu64 "\n", k);
#endif
		return;
	}
	if (it->second.dbcurrent_) {
#ifdef DEBUG_TRANS
		fprintf(stderr, "transaction::uncache current key=%" 
				PRIu64 " state %d at %p %p\n", 
			k, it->second.state_, it->second.oldcopy_, 
			it->second.newcopy_);
#endif
		if (it->second.state_ == obj_info::CLEAN) {
			it->second.shallow_dest_(it->second.oldcopy_);
			cache_.erase(it);
#ifdef DEBUG_TRANS
			fprintf(stderr, "transaction::uncache discarded "
					"the clean key=%" PRIu64 "\n", k);
#endif
		}
		return;
	}
	if (it->second.state_ != obj_info::DIRTY) {
#ifdef DEBUG_TRANS
		fprintf(stderr, "transaction::uncache ignore key=%" 
				PRIu64 " state %d at %p %p\n", 
			k, it->second.state_, it->second.oldcopy_, 
			it->second.newcopy_);
#endif
	}

	int res = writeback(it);
	if (res != 0) {
		commitfailed_ = true;
		throw trans_exception(res, "error in db put");
	}
}

int transaction::writeback(obj_map::iterator &it)
{
#ifdef DEBUG_TRANS
	fprintf(stderr, "transaction::writeback key=%" 
			PRIu64 " at %p, %zu bytes\n", 
		it->first, it->second.newcopy_, 
		it->second.objsz_);
#endif

	key_.data = (void *)&it->first;
	data_.data = it->second.newcopy_;
	data_.ulen = data_.size = it->second.objsz_;

	int res = db_->table_->put(db_->table_, txn_, 
		&key_, &data_, 0);
	if (res != 0) commitfailed_ = true;

	if (res == 0) {
		it->second.dbcurrent_ = true;
	// DEBUG {

		// debuging
#ifdef DEBUG_TRANS
		// read back the data
		data_.ulen = data_.size = it->second.objsz_;
		int res = db_->table_->get(db_->table_, txn_, 
			&key_, &data_, 0);
		if (res != 0) {
			fprintf(stderr, "get back error %s\n", 
				db_strerror(res));
		} else {
			fprintf(stderr, "get back key=%" PRIu64 
					" buffer %zu bytes, got %zu bytes\n",
				*(uint64_t *)key_.data, it->second.objsz_, 
				(size_t)data_.size);
			if (data_.size != it->second.objsz_) {
				abort();
			}
		}
#endif
	// DEBUG }
	}

	return res;
}
// END


// ---------------------------------
// a double-linked list of strings

// [[ex06cj]]
// DEBUG descr A node of a double-linked list in transactional memory.
class string_node {
public:
	tmptr<string_node> prev_;
	tmptr<string_node> next_;

public:
	// the "normal" constructors
	string_node() :
		str_(new std::string)
	{ }
	string_node(const std::string &s) :
		str_(new std::string(s))
	{ }
	string_node(const char *s) :
		str_(new std::string(s))
	{ }
	// the destructor must be empty

	// special empty constructor for extracting from
	// database
	string_node(tmptr_token &t)
	{ }
	// special functions for a complete destruction
	void destroy_tmptrs(transaction &trans)
	{ }
	void destroy()
	{
		delete str_;
	}

	// the copy constructor must do a deep copy
	string_node(const string_node &n) :
		prev_(n.prev_),
		next_(n.next_),
		str_(new std::string(n.const_str()))
	{ }

	// access the embedded string
	const std::string &const_str() const
	{
		return *str_;
	}
	std::string &str()
	{
		return *str_;
	}

protected:
	// must not contain any complex objects directly
	// but reference them
	std::string *str_;

private:
	// no assignment
	void operator=(const string_node &n);
};
// END

// [[ex06ck]]
// DEBUG descr A double-linked list in transactional memory.
class string_list
{
protected:
	tmptr<string_node> head_;

public:
	// the "normal" constructor
	string_list(transaction &trans)
	{
		string_node *hd = new string_node;
		head_.add_new(trans, hd);
		// loop the head node to itself
		hd->prev_ = head_;
		hd->next_ = head_;
	}
	// the destructor must be empty

	// special empty constructor for extracting from
	// database
	string_list(tmptr_token &t)
	{ }

	// special functions for a complete destruction
	void destroy_tmptrs(transaction &trans)
	{ 
		head_.destroy(trans);
	}
	void destroy()
	{ }

	// the default copy constructor is good enough

	// ------------- list operations ---------
	// check if the list is empty
	bool empty(transaction &trans) const
	{
		const string_node *hd = head_.get(trans);
		return (hd->prev_ == head_);
	}

	// insert at the head of the list
	void insert_head(transaction &trans, 
		tmptr<string_node> newp)
	{
		insert_after(trans, head_, newp);
	}

	// insert at the tail of the list
	void insert_tail(transaction &trans, 
		tmptr<string_node> newp)
	{
		insert_before(trans, head_, newp);
	}

	// insert newp after basep
	void insert_after(transaction &trans, 
		tmptr<string_node> basep, tmptr<string_node> newp)
	{
		string_node *base = basep.get_mod(trans);
		string_node *onext = base->next_.get_mod(trans);
		string_node *n = newp.get_mod(trans);
		n->next_ = base->next_;
		n->prev_ = basep;
		base->next_ = newp;
		onext->prev_ = newp;
	}

	// insert newp before basep
	void insert_before(transaction &trans, 
		tmptr<string_node> basep, tmptr<string_node> newp)
	{
		string_node *base = basep.get_mod(trans);
		string_node *oprev = base->prev_.get_mod(trans);
		string_node *n = newp.get_mod(trans);
		n->next_ = basep;
		n->prev_ = base->prev_;
		base->prev_ = newp;
		oprev->next_ = newp;
	}

	// remove an element from the list
	void remove(transaction &trans, 
		tmptr<string_node> nodep)
	{
		string_node *n = nodep.get_mod(trans);
		string_node *oprev = n->prev_.get_mod(trans);
		string_node *onext = n->next_.get_mod(trans);
		oprev->next_ = n->next_;
		onext->prev_ = n->prev_;
		n->prev_ = 0;
		n->next_ = 0;
	}

	// get the first element (or NULL)
	tmptr<string_node> get_first(
		transaction &trans) const
	{
		const string_node *hd = head_.get(trans);
		if (hd->prev_ == head_)
			return 0;
		else
			return hd->next_;
	}

	// get the last element (or NULL)
	tmptr<string_node> get_last(
		transaction &trans) const
	{
		const string_node *hd = head_.get(trans);
		if (hd->prev_ == head_)
			return 0;
		else
			return hd->prev_;
	}

	// return the element next from this one (or NULL)
	tmptr<string_node> get_next(transaction &trans, 
		tmptr<string_node> nodep) const
	{
		const string_node *n = nodep.get(trans);
		if (n->next_ == head_)
			return 0;
		else
			return n->next_;
	}

	// return the element previous from this one (or NULL)
	tmptr<string_node> get_prev(transaction &trans, 
		tmptr<string_node> nodep) const
	{
		const string_node *n = nodep.get(trans);
		if (n->prev_ == head_)
			return 0;
		else
			return n->prev_;
	}

private:
	// no default constructor, no assignment
	string_list();
	void operator=(const string_list &n);
};
// END

// -------------- test of linked lists -----------------

// [[ex06cl]]
// DEBUG descr Usage of a double-linked list in transactional memory.
class worker_add_node : public pw::pwthread
{
public:
	worker_add_node(pw::barrier *bar, database *db, 
			tmptr<string_list> list, bool athead, 
			int id, int count) :
		bar_(bar), db_(db), list_(list), athead_(athead), 
		id_(id), count_(count)
	{ }

	virtual void *execute()
	{
		bar_->sync();
		for (int i = 0; i < count_; i++) {
			sched_yield();
			while(true)
			{
				try {
					transaction trans(db_);

					char s[100];
					snprintf(s, sizeof(s), "%3d - %3d %c", 
						id_, i, athead_? 'f' : 't');
					tmptr<string_node> np;
					np.add_new(trans, new string_node(s));

					sched_yield();

					string_list *l = list_.get_mod(trans);
					if (athead_)
						l->insert_head(trans, np);
					else
						l->insert_tail(trans, np);

					sched_yield();

					trans.commit();
					fprintf(stderr, "%s\n", s);
				} catch(trans_exception &ex) {
					fprintf(stderr, 
						"worker_add_node %d: retry trans, %s\n", 
						id_, ex.what());
					continue;
				}
				break; // no exception, succeeded
			}
		}
		return 0;
	}

protected:
	pw::barrier *bar_;
	database *db_;
	tmptr<string_list> list_;
	bool athead_;
	int id_;
	int count_;
};

class worker_del_node : public pw::pwthread
{
public:
	worker_del_node(pw::barrier *bar, database *db, 
			tmptr<string_list> list, bool athead, 
			int id, int count) :
		bar_(bar), db_(db), list_(list), athead_(athead), 
		id_(id), count_(count)
	{ }

	virtual void *execute()
	{
		bar_->sync();
		for (int i = 0; i < count_; i++) {
			sched_yield();
			while(true)
			{
				try {
					transaction trans(db_);

					char s[100];

					sched_yield();

					string_list *l = list_.get_mod(trans);
					tmptr<string_node> np;
					if (athead_)
						np = l->get_first(trans);
					else
						np = l->get_last(trans);
					l->remove(trans, np);

					const string_node *n = np.get(trans);
					snprintf(s, sizeof(s), 
						"%3d - %3d %c <- %s", 
						id_, i, athead_? 'f' : 't',
						n->const_str().c_str());

					sched_yield();

					np.destroy(trans);

					sched_yield();

					trans.commit();
					fprintf(stderr, "%s\n", s);
				} catch(trans_exception &ex) {
					fprintf(stderr, 
						"worker_del_node %d: retry trans, %s\n", 
						id_, ex.what());
					continue;
				}
				break; // no exception, succeeded
			}
		}
		return 0;
	}

protected:
	pw::barrier *bar_;
	database *db_;
	tmptr<string_list> list_;
	bool athead_;
	int id_;
	int count_;
};

void test_string_list(database *db)
{
	const int NELEM = 100;
	const int NWORKERS = 10;
	pw::pwthread *workers[NWORKERS];
	pw::barrier bar(NWORKERS);
	tmptr<string_list> lst;

	// --- construct the list
	while(true)
	{
		try {
			transaction trans(db);
			lst.add_new(trans, new string_list(trans));
			trans.commit();
		} catch(trans_exception &ex) {
			fprintf(stderr, 
				"test_string_list create: retry trans, %s\n", 
				ex.what());
			continue;
		}
		break; // no exception, succeeded
	}

	for (int i = 0; i < NWORKERS; i++) {
		workers[i] = new worker_add_node(&bar, db, lst, 
			i%2, i, NELEM/NWORKERS);
		workers[i]->start();
	}

	bar.wait();
	fprintf(stderr, "--- start insert!\n");
	bar.signal();

	for (int i = 0; i < NWORKERS; i++) {
		workers[i]->join();
		delete workers[i];
	}

	// --- print the list
	while(true)
	{
		try {
			transaction trans(db);
			fprintf(stderr, "--- resulting list:\n");
			const string_list *lp = lst.get(trans);
			tmptr<string_node> node;
			tmptr<string_node> pnode = 0;
			for (node = lp->get_first(trans); node != 0; 
					node = lp->get_next(trans, node)) {
				if (pnode != 0)
					pnode.uncache(trans);
				const string_node *np = node.get(trans);
				fprintf(stderr, "%s\n", 
					np->const_str().c_str());
				pnode = node;
			}
			trans.rollback();
		} catch(trans_exception &ex) {
			fprintf(stderr, 
				"test_string_list print: retry trans, %s\n", 
				ex.what());
			continue;
		}
		break; // no exception, succeeded
	}

	// ---  take the list apart
	for (int i = 0; i < NWORKERS; i++) {
		workers[i] = new worker_del_node(&bar, db, lst, 
			i%2, i, NELEM/NWORKERS);
		workers[i]->start();
	}

	bar.wait();
	fprintf(stderr, "--- start removal!\n");
	bar.signal();

	for (int i = 0; i < NWORKERS; i++) {
		workers[i]->join();
		delete workers[i];
	}

	// --- delete the list
	while(true)
	{
		try {
			transaction trans(db);
			lst.destroy(trans);
			trans.commit();
		} catch(trans_exception &ex) {
			fprintf(stderr, 
				"test_string_list destroy: retry trans, %s\n", 
				ex.what());
			continue;
		}
		break; // no exception, succeeded
	}
}

// ----------- main --------------

int main()
{
	int res;
	database db;
	char dir[] = "/dev/shm/db_XXXXXX";

	// NOT mkstemp(), it will create a file, and a
	// directory is needed!
	mktemp(dir); 
	printf("%s\n", dir);
	res = db.open(dir, 1024, 1024*10, 100);
	if (res != 0) {
		fprintf(stderr, "database open error: %s\n", 
			db_strerror(res));
		exit(1);
	}

	test_string_list(&db);
	return 0;
}
// END

/* Sample output (with DB_LOCK_DEADLOCK messages filtered out):
--- start insert!
  5 -   0 f
  5 -   1 f
  5 -   2 f
  5 -   3 f
  5 -   4 f
  5 -   5 f
  5 -   6 f
  5 -   7 f
  0 -   0 t
  0 -   1 t
  0 -   2 t
  0 -   3 t
  2 -   0 t
  2 -   1 t
  2 -   2 t
  2 -   3 t
  2 -   4 t
  2 -   5 t
  2 -   6 t
  2 -   7 t
  2 -   8 t
  2 -   9 t
  3 -   0 f
  3 -   1 f
  3 -   2 f
  3 -   3 f
  3 -   4 f
  3 -   5 f
  3 -   6 f
  3 -   7 f
  3 -   8 f
  3 -   9 f
  6 -   0 t
  6 -   1 t
  6 -   2 t
  6 -   3 t
  6 -   4 t
  1 -   0 f
  1 -   1 f
  1 -   2 f
  1 -   3 f
  1 -   4 f
  6 -   5 t
  6 -   6 t
  6 -   7 t
  6 -   8 t
  6 -   9 t
  8 -   0 t
  8 -   1 t
  8 -   2 t
  8 -   3 t
  8 -   4 t
  8 -   5 t
  8 -   6 t
  8 -   7 t
  8 -   8 t
  8 -   9 t
  5 -   8 f
  5 -   9 f
  4 -   0 t
  4 -   1 t
  4 -   2 t
  4 -   3 t
  4 -   4 t
  4 -   5 t
  4 -   6 t
  4 -   7 t
  4 -   8 t
  4 -   9 t
  9 -   0 f
  9 -   1 f
  9 -   2 f
  9 -   3 f
  9 -   4 f
  7 -   0 f
  7 -   1 f
  7 -   2 f
  7 -   3 f
  7 -   4 f
  7 -   5 f
  7 -   6 f
  7 -   7 f
  7 -   8 f
  7 -   9 f
  0 -   4 t
  0 -   5 t
  0 -   6 t
  0 -   7 t
  0 -   8 t
  0 -   9 t
  9 -   5 f
  9 -   6 f
  9 -   7 f
  9 -   8 f
  9 -   9 f
  1 -   5 f
  1 -   6 f
  1 -   7 f
  1 -   8 f
  1 -   9 f
--- resulting list:
  1 -   9 f
  1 -   8 f
  1 -   7 f
  1 -   6 f
  1 -   5 f
  9 -   9 f
  9 -   8 f
  9 -   7 f
  9 -   6 f
  9 -   5 f
  7 -   9 f
  7 -   8 f
  7 -   7 f
  7 -   6 f
  7 -   5 f
  7 -   4 f
  7 -   3 f
  7 -   2 f
  7 -   1 f
  7 -   0 f
  9 -   4 f
  9 -   3 f
  9 -   2 f
  9 -   1 f
  9 -   0 f
  5 -   9 f
  5 -   8 f
  1 -   4 f
  1 -   3 f
  1 -   2 f
  1 -   1 f
  1 -   0 f
  3 -   9 f
  3 -   8 f
  3 -   7 f
  3 -   6 f
  3 -   5 f
  3 -   4 f
  3 -   3 f
  3 -   2 f
  3 -   1 f
  3 -   0 f
  5 -   7 f
  5 -   6 f
  5 -   5 f
  5 -   4 f
  5 -   3 f
  5 -   2 f
  5 -   1 f
  5 -   0 f
  0 -   0 t
  0 -   1 t
  0 -   2 t
  0 -   3 t
  2 -   0 t
  2 -   1 t
  2 -   2 t
  2 -   3 t
  2 -   4 t
  2 -   5 t
  2 -   6 t
  2 -   7 t
  2 -   8 t
  2 -   9 t
  6 -   0 t
  6 -   1 t
  6 -   2 t
  6 -   3 t
  6 -   4 t
  6 -   5 t
  6 -   6 t
  6 -   7 t
  6 -   8 t
  6 -   9 t
  8 -   0 t
  8 -   1 t
  8 -   2 t
  8 -   3 t
  8 -   4 t
  8 -   5 t
  8 -   6 t
  8 -   7 t
  8 -   8 t
  8 -   9 t
  4 -   0 t
  4 -   1 t
  4 -   2 t
  4 -   3 t
  4 -   4 t
  4 -   5 t
  4 -   6 t
  4 -   7 t
  4 -   8 t
  4 -   9 t
  0 -   4 t
  0 -   5 t
  0 -   6 t
  0 -   7 t
  0 -   8 t
  0 -   9 t
--- start removal!
  7 -   0 f <-   1 -   9 f
  7 -   1 f <-   1 -   8 f
  7 -   2 f <-   1 -   7 f
  0 -   0 t <-   0 -   9 t
  7 -   3 f <-   1 -   6 f
  8 -   0 t <-   0 -   8 t
  0 -   1 t <-   0 -   7 t
  8 -   1 t <-   0 -   6 t
  8 -   2 t <-   0 -   5 t
  0 -   2 t <-   0 -   4 t
  5 -   0 f <-   1 -   5 f
  8 -   3 t <-   4 -   9 t
  8 -   4 t <-   4 -   8 t
  0 -   3 t <-   4 -   7 t
  8 -   5 t <-   4 -   6 t
  0 -   4 t <-   4 -   5 t
  8 -   6 t <-   4 -   4 t
  0 -   5 t <-   4 -   3 t
  0 -   6 t <-   4 -   2 t
  8 -   7 t <-   4 -   1 t
  8 -   8 t <-   4 -   0 t
  0 -   7 t <-   8 -   9 t
  0 -   8 t <-   8 -   8 t
  7 -   4 f <-   9 -   9 f
  5 -   1 f <-   9 -   7 f
  1 -   0 f <-   9 -   8 f
  4 -   0 t <-   8 -   7 t
  4 -   1 t <-   8 -   6 t
  4 -   2 t <-   8 -   5 t
  4 -   3 t <-   8 -   4 t
  4 -   4 t <-   8 -   3 t
  4 -   5 t <-   8 -   2 t
  4 -   6 t <-   8 -   1 t
  4 -   7 t <-   8 -   0 t
  4 -   8 t <-   6 -   9 t
  4 -   9 t <-   6 -   8 t
  7 -   5 f <-   9 -   6 f
  7 -   6 f <-   9 -   5 f
  7 -   7 f <-   7 -   9 f
  7 -   8 f <-   7 -   8 f
  7 -   9 f <-   7 -   7 f
  9 -   0 f <-   7 -   6 f
  9 -   1 f <-   7 -   5 f
  9 -   2 f <-   7 -   4 f
  9 -   3 f <-   7 -   3 f
  9 -   4 f <-   7 -   2 f
  9 -   5 f <-   7 -   1 f
  9 -   6 f <-   7 -   0 f
  9 -   7 f <-   9 -   4 f
  9 -   8 f <-   9 -   3 f
  9 -   9 f <-   9 -   2 f
  3 -   0 f <-   9 -   1 f
  3 -   1 f <-   9 -   0 f
  3 -   2 f <-   5 -   9 f
  3 -   3 f <-   5 -   8 f
  3 -   4 f <-   1 -   4 f
  3 -   5 f <-   1 -   3 f
  3 -   6 f <-   1 -   2 f
  3 -   7 f <-   1 -   1 f
  5 -   2 f <-   1 -   0 f
  3 -   8 f <-   3 -   9 f
  5 -   3 f <-   3 -   8 f
  3 -   9 f <-   3 -   7 f
  5 -   4 f <-   3 -   6 f
  5 -   5 f <-   3 -   5 f
  5 -   6 f <-   3 -   4 f
  5 -   7 f <-   3 -   3 f
  5 -   8 f <-   3 -   2 f
  1 -   1 f <-   3 -   1 f
  5 -   9 f <-   3 -   0 f
  1 -   2 f <-   5 -   7 f
  1 -   3 f <-   5 -   6 f
  1 -   4 f <-   5 -   5 f
  1 -   5 f <-   5 -   4 f
  1 -   6 f <-   5 -   3 f
  1 -   7 f <-   5 -   2 f
  1 -   8 f <-   5 -   1 f
  1 -   9 f <-   5 -   0 f
  8 -   9 t <-   6 -   7 t
  0 -   9 t <-   6 -   6 t
  6 -   0 t <-   6 -   5 t
  6 -   1 t <-   6 -   4 t
  6 -   2 t <-   6 -   3 t
  6 -   3 t <-   6 -   2 t
  6 -   4 t <-   6 -   1 t
  6 -   5 t <-   6 -   0 t
  6 -   6 t <-   2 -   9 t
  6 -   7 t <-   2 -   8 t
  6 -   8 t <-   2 -   7 t
  6 -   9 t <-   2 -   6 t
  2 -   0 t <-   2 -   5 t
  2 -   1 t <-   2 -   4 t
  2 -   2 t <-   2 -   3 t
  2 -   3 t <-   2 -   2 t
  2 -   4 t <-   2 -   1 t
  2 -   5 t <-   2 -   0 t
  2 -   6 t <-   0 -   3 t
  2 -   7 t <-   0 -   2 t
  2 -   8 t <-   0 -   1 t
  2 -   9 t <-   0 -   0 t
/dev/shm/db_Qd25aS
*/
